package cn.edu360.report

import cn.edu360.utils.ConfigHandler
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

object DataLabel {
  def main(args: Array[String]): Unit = {
    // 1) 广告位类型（标签格式： LC03->1 或者 LC16->1）xx 为数字，小于 10 补 0，把广告 位类型名称，LN 插屏->1
    // 2) App 名称（标签格式： APPxxxx->1）xxxx 为 App 名称，使用缓存文件 appname_dict 进行名称转换； APP 爱奇艺->1
    // 3) 渠道（标签格式： CNxxxx->1）xxxx 为渠道 ID(adplatformproviderid)
    // 4) 设备： a) (操作系统 -> 1) b) (联网方 -> 1) c) (运营商 -> 1)
    // 5) 关键字（标签格式：Kxxx->1）xxx 为关键字，关键字个数不能少于 3 个字符，且不能 超过 8 个字符；关键字中如包含‘‘|’’，则分割成数组，转化成多个关键字标签
    // 6) 地域标签（省标签格式：ZPxxx->1, 地市标签格式: ZCxxx->1）xxx 为省或市名称
    //33 adspacetype: Int, 广告位类型（1：banner 2：插屏 3：全屏）
    //34 adspacetypename: String, 广告位类型名称（banner、插屏、全屏）
    //15 appname: String, 应用名称
    //5 adplatformproviderid: Int, 广告平台商 id (>= 100000: rtb)渠道 ID
    //84 keywords: String, 关键字
    //25 provincename: String, 设备所在省份名称
    //26 cityname: String, 设备所在城市名称
    //18 client: Int, 设备类型 （1：android 2：ios 3：wp）
    //30 networkmannername: String,联网方式名称
    //28 ispname: String, 运营商名称
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("数据标签化")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)

    /*val startData: RDD[(Int, String, String, Int, Int, String, String, String, String, String)] = rawDataFrame.map(t => {
      //广告位类型
      val adType = t.getAs[Int]("adspacetype")
      //广告位类型名称
      val adTypeName = t.getAs[String]("adspacetypename")
      //App名称
      val appName = t.getAs[String]("appname")
      //渠道 ID
      val adProviderId = t.getAs[Int]("adplatformproviderid")
      //设备类型
      val client = t.getAs[Int]("client")
      //联网方式名称
      val netName = t.getAs[String]("networkmannername")
      //运营商名称
      val ispname = t.getAs[String]("ispname")
      //keywords
      val keywords = t.getAs[String]("keywords")
      //设备所在省份名称
      val pname = t.getAs[String]("provincename")
      //设备所在城市名称
      val cname = t.getAs[String]("cityname")
      (adType, adTypeName, appName, adProviderId, client, netName, ispname, keywords, pname, cname)
    })*/
    /* startData.map(t =>{
       if(t._1==1) "LC01" else t._1
       else if (t._1==)
     })*/
    // 字典数据广播出去
    val appdictMap: Map[String, String] = sc.textFile(ConfigHandler.appDictPath)
      .map(line => line.split("\t", -1))
      .filter(_.length >= 5)
      .map(t => (t(4), t(1))).collect().toMap
    val broadcast: Broadcast[Map[String, String]] = sc.broadcast(appdictMap)
    val sQLContext = new SQLContext(sc)
    // 读取数据
    val rawDataFrame = sQLContext.read.parquet(ConfigHandler.parquetPath)
    import sQLContext.implicits._
    rawDataFrame.filter("appid!='0' or appname!='0'")
      .filter("keywords>=3 and keywords<=8")
      .map(t => {
        val imei = t.getAs[String]("imei")
        val mac = t.getAs[String]("mac")
        val idfa = t.getAs[String]("idfa")
        val openudid = t.getAs[String]("openudid")
        val androidid = t.getAs[String]("androidid")

        val imeimd5 = t.getAs[String]("imeimd5")
        val macmd5 = t.getAs[String]("macmd5")
        val idfamd5 = t.getAs[String]("idfamd5")
        val openudidmd5 = t.getAs[String]("openudidmd5")
        val androididmd5 = t.getAs[String]("androididmd5")

        val imeisha1 = t.getAs[String]("imeisha1")
        val macsha1 = t.getAs[String]("macsha1")
        val idfasha1 = t.getAs[String]("idfasha1")
        val openudidsha1 = t.getAs[String]("openudidsha1")
        val androididsha1 = t.getAs[String]("androididsha1")
        val id: String = if (imei != "0") imei
        else if (mac != "0") mac
        else if (idfa != "0") idfa
        else if (openudid != "0") openudid
        else if (androidid != "0") androidid
        else if (imeimd5 != "0") imeimd5
        else if (macmd5 != "0") macmd5
        else if (idfamd5 != "0") idfamd5
        else if (openudidmd5 != "0") openudidmd5
        else if (androididmd5 != "0") androididmd5
        else if (imeisha1 != "0") imeisha1
        else if (macsha1 != "0") macsha1
        else if (idfasha1 != "0") idfasha1
        else if (openudidsha1 != "0") openudidsha1
        else if (androididsha1 != "0") androididsha1
        else "null"
        //33 adspacetype: Int, 广告位类型（1：banner 2：插屏 3：全屏）
        //广告位类型（标签格式： LC03->1 或者 LC16->1）xx 为数字，小于 10 补 0，把广告 位类型名称，LN 插屏->1
        val adspacetype = t.getAs[Int]("adspacetype")
        val adspacetype1 = if (adspacetype < 10) s"LC0$adspacetype" else s"LC$adspacetype"
        val adspacetypename = t.getAs[String]("adspacetypename")
        val adspacetypename1 = s"LN$adspacetypename"
        // App 名称（标签格式： APPxxxx->1）xxxx 为 App 名称，使用缓存文件 appname_dict 进行名称转换；
        //获取 app名称
        val appname = t.getAs[String]("appname")
        val appid = t.getAs[String]("appid")
        val appname1 = if (appname == "0") {
          broadcast.value.getOrElse(appid, appid)
        } else appname
        // 渠道（标签格式： CNxxxx->1）xxxx 为渠道 ID(adplatformproviderid)
        val adplatformproviderid = t.getAs[Int]("adplatformproviderid")
        val adplatformproviderid1 = s"CN$adplatformproviderid"
        //操作系统
        //设备操作系统 1 Android D00010001 2 IOS D00010002 3 WinPhone D00010003 _ 其他 D0001000
        val client = t.getAs[Int]("client")
        val client1 = if (client == 1) "Android D00010001"
        else if (client == 2) "IOS D00010002"
        else if (client == 3) "WinPhone D00010003"
        else "_ 其他 D0001000"
        //联网方式
        //WIFI D00020001 4G D00020002 3G D00020003 2G D00020004 _ D00020005
        val networkmannername = t.getAs[String]("networkmannername")
        val networkmannername1 = if (networkmannername.equals("Wifi")) "WIFI D00020001"
        else if (networkmannername.equals("4G")) "4G D00020002"
        else if (networkmannername.equals("3G")) "3G D00020003"
        else if (networkmannername.equals("2G")) "2G D00020004"
        else "_ D00020005"
        //设备运营商方式
        //移动 D00030001 联通 D00030002 电信 D00030003 _ D00030004
        val ispname = t.getAs[String]("ispname")
        val ispname1 = if (ispname.equals("移动")) "移动 D00030001"
        else if (ispname.equals("联通")) "联通 D00030002"
        else if (ispname.equals("电信")) "电信 D00030003"
        else "_ D00030004"
        // 关键字（标签格式：Kxxx->1）xxx 为关键字
        //关键字中如包含‘‘|’’，则分割成数组，转化成多个关键字标签
        val keywords = t.getAs[String]("keywords")
        val keywords1 = if (keywords.contains("[|]")) {
          val strings: Array[String] = keywords.split("[|]")
          val str: String = strings.mkString("K", ",", "")
          str
        } else keywords
        //地域标签（省标签格式：ZPxxx->1, 地市标签格式: ZCxxx->1）xxx 为省或市名称
        val provincename: String = t.getAs[String]("provincename")
        val provincename1 = s"ZP$provincename"
        val cityname = t.getAs[String]("cityname")
        val cityname1 = s"ZC$cityname"
        (id, (adspacetypename1, appname1, adplatformproviderid1, client1,
          networkmannername1, ispname1, keywords1, provincename1, cityname1))
      }).foreach(println)
  }
}


